package api

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/go-caipu-team/go-caipu-plugin-sdk/backend"
	"github.com/gin-gonic/gin"
	"go-caipu/pkg/api/helper"
	"go-caipu/pkg/plugins/backendplugin"
	"go-caipu/pkg/services/admin/sysuser"
	"go-caipu/pkg/util"
	"go-caipu/pkg/util/proxyutil"
	"io"
	"net/http"
	"net/textproto"
	"net/url"
	"strings"
	"sync"
)

const (
	NO        int = 0
	BasicUser int = 1
	ALLUser   int = 2 // Green 的值是 1 (iota 自动递增)

)

var (
	extendUserIDKey   = "x-extendUserId"
	extendUserNameKey = "x-extendUserName"
	extendRoleIdKey   = "x-extendRoleId"
	extendDeptIdKey   = "x-extendDeptId"
)

// CallResource 调用后端资源
func (hs *HTTPServer) CallResource(c *gin.Context) {
	hs.callPluginResource(c, c.Param("pluginId"))
}

// AsyncCallResource 异步调用插件资源
func (hs *HTTPServer) AsyncCallResource(c *gin.Context) {
	pluginID := c.Param("pluginId")
	pCtx, found, err := hs.pluginContextProvider.Get(c.Request.Context(), pluginID)
	if err != nil {
		c.JSON(500, fmt.Sprintf("pluginID:%s,  Failed to get plugin settings", pluginID))
		return
	}
	if !found {
		c.JSON(404, fmt.Sprintf("Plugin( %s ) not found", pluginID))
		return
	}
	//调用地址限制 前台不能调用后端管理员插件,后端接口受鉴权控制暂时不用做控制
	p, exists := hs.pluginStore.Plugin(c.Request.Context(), pluginID)
	if exists {
		c.JSON(500, fmt.Sprintf("pluginID:%s,  Failed to get plugin settings", c.Param("pluginId")))
		return
	}
	if strings.HasPrefix(c.Request.RequestURI, "/site/") && p.BackendAdmin {
		c.JSON(403, fmt.Sprintf("Plugin( %s ) 拒绝调用", pluginID))
		return
	}

	crReq := &backend.ExecuteRequest{
		PluginContext: pCtx,
		JSON: []byte(`{
    "MessageID": "20242343",
    "To": "78055551@qq.com",
    "Cc": "78055551@qq.com",
    "Bcc": "",
    "Subject": "test",
    "Body": " 这是一个测试邮件",
    "ContentType": "text/plain",
    "Attempts": 3,
    "CallbackURL": "http://www.baidu.com"
}`),
	}

	result, err := hs.pluginClient.Execute(c.Request.Context(), crReq)

	//result, err := hs.pluginClient.Execute(c.Request.Context(), &backend.ExecuteRequest{PluginContext: pCtx})
	if err != nil {
		c.JSON(500, fmt.Sprintf("pluginID:%s,  Failed to call plugin resource", pluginID))
		return
	}

	helper.SuccessWithData(c, result)
}
func (hs *HTTPServer) callPluginResource(c *gin.Context, pluginID string) {
	pCtx, found, err := hs.pluginContextProvider.Get(c.Request.Context(), pluginID)
	if err != nil {
		c.JSON(500, fmt.Sprintf("pluginID:%s,  Failed to get plugin settings", pluginID))
		return
	}
	if !found {
		c.JSON(404, fmt.Sprintf("Plugin( %s ) not found", pluginID))
		return
	}

	//调用地址限制 前台不能调用后端管理员插件,后端接口受鉴权控制暂时不用做控制
	p, exists := hs.pluginStore.Plugin(c.Request.Context(), pluginID)
	if !exists {
		c.JSON(500, fmt.Sprintf("pluginID:%s,  get plugin data failed", c.Param("pluginId")))
		return
	}
	if strings.HasPrefix(c.Request.RequestURI, "/site/") && p.BackendAdmin {
		c.JSON(403, fmt.Sprintf("Plugin( %s ) 拒绝调用", pluginID))
		return
	}

	req, err := hs.pluginResourceRequest(c)
	if err != nil {
		c.JSON(http.StatusBadRequest, fmt.Sprintf("Failed for create plugin resources request %w", err))
		return
	}
	switch p.XExtendUser {
	case true:
		userId, err := getUserIdByToken(hs, c)
		if err != nil {
			c.JSON(401, fmt.Sprintf("Plugin( %s ) GetByHash User err:%s", pluginID, err.Error()))
			return
		}
		cmd := sysuser.GetUserCommand{UserId: userId}
		user, err := hs.userService.GetUser(c.Request.Context(), cmd)
		deptIds, _ := hs.deptService.GetDeptAndChildrenIds(c.Request.Context(), user.DeptId)

		req.Header.Set(extendUserIDKey, util.Int64ToString(user.UserId))
		req.Header.Set(extendUserNameKey, user.NickName)
		req.Header.Set(extendRoleIdKey, util.ArrToString(user.RoleIds))
		req.Header.Set(extendDeptIdKey, util.Int64ArrayToString(deptIds))
	}
	if err = hs.makePluginResourceRequest(c.Writer, req, pCtx); err != nil {
		handleCallResourceError(c, err)
	}
}

// pluginResourceRequest 接收插件资源请求
func (hs *HTTPServer) pluginResourceRequest(c *gin.Context) (*http.Request, error) {
	cloneReq := c.Request.Clone(c.Request.Context())
	rawURL := c.Param("url")
	if cloneReq.URL.RawQuery != "" {
		rawURL += "?" + cloneReq.URL.RawQuery
	}
	urlPath, err := url.Parse(rawURL)
	if err != nil {
		return nil, err
	}
	cloneReq.URL = urlPath
	return cloneReq, nil
}
func (hs *HTTPServer) makePluginResourceRequest(w http.ResponseWriter, req *http.Request, pCtx backend.PluginContext) error {
	proxyutil.PrepareProxyRequest(req)

	body, err := io.ReadAll(req.Body)
	if err != nil {
		return fmt.Errorf("failed to read request body: %w", err)
	}
	crReq := &backend.CallResourceRequest{
		PluginContext: pCtx,
		Path:          req.URL.Path,
		Method:        req.Method,
		URL:           req.URL.String(),
		Headers:       req.Header,
		Body:          body,
	}
	childCtx, cancel := context.WithCancel(req.Context())
	defer cancel()
	stream := newCallResourceResponseStream(childCtx)

	var wg sync.WaitGroup
	wg.Add(1)
	defer func() {
		if err := stream.Close(); err != nil {
			hs.log.Warn("Failed to close plugin resources stream", "err", err)
		}
		wg.Wait()
	}()

	var flushStreamErr error

	go func() {
		flushStreamErr = hs.flushStream(req.Context(), crReq, stream, w)
		wg.Done()
	}()
	if err := hs.pluginClient.CallResource(req.Context(), crReq, stream); err != nil {
		return err
	}
	return flushStreamErr
}
func (hs *HTTPServer) flushStream(ctx context.Context, req *backend.CallResourceRequest, stream callResourceClientResponseStream, w http.ResponseWriter) error {
	processedStreams := 0
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	for {
		resp, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			if processedStreams == 0 {
				return errors.New("received empty resources response")
			}
			return nil
		}
		if err != nil {
			if processedStreams == 0 {
				return fmt.Errorf("%v: %w", "failed to receive response from resources call", err)
			}

			hs.log.Error("Failed to receive response from resources call", "err", err)
			return stream.Close()
		}

		// Expected that headers and status are only part of first stream
		if processedStreams == 0 {
			var hasContentType bool
			for k, values := range resp.Headers {
				// Convert the keys to the canonical format of MIME headers.
				// This ensures that we can safely add/overwrite headers
				// even if the plugin returns them in non-canonical format
				// and be sure they won't be present multiple times in the response.
				k = textproto.CanonicalMIMEHeaderKey(k)

				switch k {
				case "Set-Cookie":
					// Due to security reasons we don't want to forward
					// cookies from a backend plugin to clients/browsers.
					continue
				case "Content-Type":
					hasContentType = true
				}

				for _, v := range values {
					// TODO: Figure out if we should use Set here instead
					// nolint:gocritic
					w.Header().Add(k, v)
				}
			}

			// Make sure a content type always is returned in response
			if !hasContentType && resp.Status != http.StatusNoContent {
				w.Header().Set("Content-Type", "application/json")
			}

			proxyutil.SetProxyResponseHeaders(w.Header())

			w.WriteHeader(resp.Status)
		}

		if _, err := w.Write(resp.Body); err != nil {
			hs.log.Error("Failed to write resources response", "err", err)
		}
		//else if hs.Features.IsEnabled(featuremgmt.FlagUseCachingService) {
		//	// Placing the new service implementation behind a feature flag until it is known to be stable
		//
		//	// The enterprise implementation of this function will use the headers and status of the first response,
		//	// And append the body of any subsequent responses. It waits for the context to be canceled before caching the cumulative result.
		//	hs.cachingService.CacheResourceResponse(ctx, req, resp)
		//}

		if flusher, ok := w.(http.Flusher); ok {
			flusher.Flush()
		}
		processedStreams++
	}
}
func handleCallResourceError(c *gin.Context, err error) {
	if errors.Is(err, backendplugin.ErrPluginUnavailable) {
		c.JSON(503, fmt.Sprintf("Plugin unavailable : %w", err))
		return
	}

	if errors.Is(err, backendplugin.ErrMethodNotImplemented) {

		c.JSON(404, fmt.Sprintf("%w Not found", err))
		return
	}
	c.JSON(500, fmt.Sprintf("Failed to call resources : %w", err))
}

// callResourceClientResponseStream is used for receiving resource call responses.
type callResourceClientResponseStream interface {
	Recv() (*backend.CallResourceResponse, error)
	Close() error
}
type callResourceResponseStream struct {
	ctx    context.Context
	stream chan *backend.CallResourceResponse
	closed bool
}

func newCallResourceResponseStream(ctx context.Context) *callResourceResponseStream {
	return &callResourceResponseStream{
		ctx:    ctx,
		stream: make(chan *backend.CallResourceResponse),
	}
}

func (s *callResourceResponseStream) Send(res *backend.CallResourceResponse) error {
	if s.closed {
		return errors.New("cannot send to a closed stream")
	}

	select {
	case <-s.ctx.Done():
		return errors.New("cancelled")
	case s.stream <- res:
		return nil
	}
}

func (s *callResourceResponseStream) Recv() (*backend.CallResourceResponse, error) {
	select {
	case <-s.ctx.Done():
		return nil, s.ctx.Err()
	case res, ok := <-s.stream:
		if !ok {
			return nil, io.EOF
		}
		return res, nil
	}
}

func (s *callResourceResponseStream) Close() error {
	if s.closed {
		return errors.New("cannot close a closed stream")
	}

	close(s.stream)
	s.closed = true
	return nil
}
